package com.tang.configuration;

import com.tang.constant.Cons;
import com.tang.service.MsgSendConfirmCallBack;
import com.tang.service.MsgSendReturnCallback;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;

import javax.annotation.Resource;

/**
 * RabbitMQ遵循了AMQP规范，用消息确认机制来保证，只要消息发送，就能确保被消费者消费来做到了消息最终一致性
 * 亦可以参考https://segmentfault.com/a/1190000012762869这个分布式事务相关的说明
 * rabbimq的简单配置与使用可以参考：https://blog.csdn.net/zhuzhezhuzhe1/article/details/80642032这个人的博客（RabbitMQ实战篇7、8、9、10、11）
 */
@Configuration
public class RabbitConfig {

//    private static final Logger logger = LogManager.getLogger(RabbitConfig.class);

//    @Value("${spring.rabbitmq.host}")
//    private String addresses;
//
//    @Value("${spring.rabbitmq.port}")
//    private String port;
//
//    @Value("${spring.rabbitmq.username}")
//    private String username;
//
//    @Value("${spring.rabbitmq.password}")
//    private String password;

//    @Value("${spring.rabbitmq.virtual-host}")
//    private String virtualHost;

//    @Value("${spring.rabbitmq.publisher-confirms}")
//    private boolean publisherConfirms;
//
//    @Value("${spring.rabbitmq.publisher-returns}")
//    private boolean publisherReturns;

//    @Value("${spring.rabbitmq.queues}")
//    private String queues;

    @Resource
    private ConnectionFactory connectionFactory;

//    @Bean
//    public ConnectionFactory connectionFactory() {
//        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
//        connectionFactory.setAddresses(addresses + ":" + port);
//        connectionFactory.setUsername(username);
//        connectionFactory.setPassword(password);
//        // connectionFactory.setVirtualHost(virtualHost);
//        // 自动读取application.yml文件中的rabbitmq相关配置后，
//        // 会自动创建ConnectionFactory以及RabbitTemplate对应Bean，
//        // 自动创建的ConnectionFactory无法完成事件的回调，如果要进行消息回调，
//        // 则这里必须要设置为true
//        connectionFactory.setPublisherConfirms(publisherConfirms);
//        connectionFactory.setPublisherReturns(publisherReturns);
//        return connectionFactory;
//    }

    /**
     * 如果想将消息进行持久化，只需要将交换机和队列持久化就可以了
     * 1. 设置交换机类型
     * 2. 将队列绑定到交换机
     * FanoutExchange: 将消息分发到所有的绑定队列，无routingkey的概念
     * HeadersExchange ：通过添加属性key-value匹配
     * DirectExchange:按照routingkey分发到指定队列
     * TopicExchange:多关键字匹配
     */
    @Bean
    TopicExchange topicExchange() {
        return new TopicExchange(Cons.TOPIC_EXCHANGE_NAME);
    }

//    @Bean
//    DirectExchange directExchange() {
//        return new DirectExchange(DIRECT_EXCHANGE_NAME, true, false);
//    }

    /**
     * 如果想将消息进行持久化，只需要将交换机和队列持久化就可以了
     * 当我们在创建队列时指定durable = true，当服务重启的时候这个队列将会存活。也就是说队列被持久化了。
     * 同样的，交换机的durable也和队列同理
     * durable="true" 持久化消息队列 ， rabbitmq重启的时候不需要创建新的队列
     * auto-delete 表示消息队列没有在使用时将被自动删除 默认是false
     * exclusive  表示该消息队列是否只在当前connection生效,默认是false
     */
    @Bean
    public Queue topicBasicQueue() {
//        Map<String, Object> arguments = new HashMap<>();
//        arguments.put("x-message-ttl", 60000);//60秒自动删除
//        return new Queue(QUEUE_NAME3,true,false,true,arguments);
        return new Queue(Cons.TOPIC_QUEUE_NAME_BASIC, true); // 队列持久化
    }

    @Bean
    public Queue topicObjectQueue() {
        return new Queue(Cons.TOPIC_QUEUE_NAME_OBJECT, true); // 队列持久化
    }

//    @Bean
//    public Queue directQueue() {
//        return new Queue(DIRECT_QUEUE_NAME, true, false, false);
//    }

    @Bean
    public Binding topicBasicBinding() {
        return BindingBuilder.bind(topicBasicQueue()).to(topicExchange()).with(Cons.TOPIC_ROUTING_KEY_BASIC);
    }

    @Bean
    public Binding topicObjectBinding() {
        return BindingBuilder.bind(topicObjectQueue()).to(topicExchange()).with(Cons.TOPIC_ROUTING_KEY_OBJECT);
    }

//    @Bean
//    public Binding directBinding() {
//        return BindingBuilder.bind(directQueue()).to(directExchange()).with(DIRECT_ROUTING_KEY);
//    }

    /**
     * 定义rabbit template用于数据的接收和发送,可以设置消息确认机制和回调:
     * 在生产者需要消息发送后的回调，需要对rabbitTemplate设置ConfirmCallback对象，
     * 由于不同的生产者需要对应不同的ConfirmCallback，如果rabbitTemplate设置为单例bean，则所有的rabbitTemplate
     * 实际的ConfirmCallback为最后一次申明的ConfirmCallback。
     * 通过使用RabbitTemplate来对开发者提供API操作, 因为要设置回调类，所以应是prototype类型，
     * 如果是singleton类型，则回调类为最后一次设置
     *
     * @return
     */
    @Bean
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        // template.setMessageConverter(); 可以自定义消息转换器  默认使用的JDK的，所以消息对象需要实现Serializable
        // template.setMessageConverter(new Jackson2JsonMessageConverter());

        // 若使用confirm-callback或return-callback，
        // 必须要配置publisherConfirms或publisherReturns为true
        // 每个rabbitTemplate只能有一个confirm-callback和return-callback
        template.setConfirmCallback(msgSendConfirmCallBack());

        // 使用return-callback时必须设置mandatory为true，或者在配置中设置mandatory-expression的值为true，
        // 可针对每次请求的消息去确定’mandatory’的boolean值，
        // 只能在提供’return -callback’时使用，与mandatory互斥
        template.setReturnCallback(msgSendReturnCallback());
        template.setMandatory(true);
        return template;
    }

    // 关于 msgSendConfirmCallBack 和 msgSendReturnCallback 的回调说明：
    // 如果消息没有到exchange,则confirm回调,ack=false
    // 如果消息到达exchange,则confirm回调,ack=true
    // exchange到queue成功,则不回调return
    // exchange到queue失败,则回调return(需设置mandatory=true,否则不回调,消息就丢了)

    // 消息确认机制:
    // Confirms给客户端一种轻量级的方式，能够跟踪哪些消息被broker处理，
    // 哪些可能因为broker宕掉或者网络失败的情况而重新发布。
    // 确认并且保证消息被送达，提供了两种方式：发布确认和事务。(两者不可同时使用)
    // 在channel为事务时，不可引入确认模式；同样channel为确认模式下，不可使用事务。
    @Bean
    public MsgSendConfirmCallBack msgSendConfirmCallBack() {
        return new MsgSendConfirmCallBack();
    }

    @Bean
    public MsgSendReturnCallback msgSendReturnCallback() {
        return new MsgSendReturnCallback();
    }

    /**
     * 控制消息的接收：
     *    消息监听器容器是一个用于查看JMS目标等待消息到达的特殊bean，
     *    一旦消息到达它就可以获取到消息，并通过调用onMessage()方法将消息传递给一个MessageListener实现
     *    SimpleMessageListenerContainer：最简单的消息监听器容器，只能处理固定数量的JMS会话，且不支持事务
     *    DefaultMessageListenerContainer：这个消息监听器容器建立在SimpleMessageListenerContainer容器之上，添加了对事务的支持
     *    serversession.ServerSessionMessage.ListenerContainer：这是功能最强大的消息监听器，与DefaultMessageListenerContainer相同，它支持事务，但是它还允许动态地管理JMS会话。
     *    在使用消息监听器容器时一定要将自定义的消息监听器置入到容器中，这样才可以在收到信息时，容器把消息转向相应的消息监听器进行处理。
     *    需要注意的是，应该只是在Destination为Queue的时候才使用多个MessageConsumer（Queue中的一个消息只能被一个Consumer接收），虽
     *    然使用多个MessageConsumer会提高消息处理的性能，但是消息处理的顺序却得不到保证。消息被接收的顺序仍然是消息发送时的顺序，
     *    但是由于消息可能会被并发处理，因此消息处理的顺序可能和消息发送的顺序不同。
     *    此外，不应该在Destination为Topic的时候使用多个MessageConsumer，因为多个MessageConsumer会接收到同样的消息
     * @return
     */
//    @Bean
//    public SimpleMessageListenerContainer messageContainer() {
//		/*Queue[] q = new Queue[queues.split(",").length];
//		for (int i = 0; i < queues.split(",").length; i++) {
//			q[i] = new Queue(queues.split(",")[i]);
//		}*/
//        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
//        container.setQueues(topicBasicQueue());
//        container.setExposeListenerChannel(true);
//        container.setMaxConcurrentConsumers(1);
//        container.setConcurrentConsumers(1);
//        container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //设置确认模式手工确认
//        container.setMessageListener(new ChannelAwareMessageListener() {
//            public void onMessage(Message message, com.rabbitmq.client.Channel channel) throws Exception {
//                try {
//                    logger.info("消费端接收到消息:>>>>>>>>>>>" + message.getMessageProperties() + ":" + new String(message.getBody(), "UTF-8"));
//                    logger.info("topic:>>>>>>>>>" + message.getMessageProperties().getReceivedRoutingKey());
//                    // deliveryTag是消息传送的次数，我这里是为了让消息队列的第一个消息到达的时候抛出异常，处理异常让消息重新回到队列，然后再次抛出异常，处理异常拒绝让消息重回队列
//					/*if (message.getMessageProperties().getDeliveryTag() == 1
//							|| message.getMessageProperties().getDeliveryTag() == 2) {
//						throw new Exception();
//					}*/
//                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); // false只确认当前一个消息收到，true确认所有consumer获得的消息
//                } catch (Exception e) {
//                    e.printStackTrace();
//                    if (message.getMessageProperties().getRedelivered()) {
//                        logger.info("消息已重复处理失败,拒绝再次接收...");
//                        channel.basicReject(message.getMessageProperties().getDeliveryTag(), true); // 拒绝消息
//                    } else {
//                        logger.info("消息即将再次返回队列处理...");
//                        channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true); // requeue为是否重新回到队列
//                    }
//                }
//            }
//        });
//        return container;
//    }

}
